/* * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * */
/* Copyright 2002-2003 by Mark Wolgemuth, All rights reserved.             */
/* Copyright 2002-2003 by Employease, Inc.  All rights reserved.           */
/*                                                                         */
/* Redistribution and use in source and binary forms, with or without      */
/* modification, are permitted provided that the following conditions      */
/* are met:                                                                */
/*                                                                         */
/*    1. Redistributions of source code must retain the above copyright    */
/* notice, this list of conditions and the following disclaimer.           */
/*    2. Redistributions in binary form must reproduce the above copyright */
/* notice, this list of conditions and the following disclaimer in the     */
/* documentation and/or other materials provided with the distribution.    */
/*                                                                         */
/* THIS SOFTWARE IS PROVIDED BY THE MOD_ATHENA PROJECT ``AS IS'' AND ANY   */
/* EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE       */
/* IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR      */
/* PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE FREEBSD PROJECT OR        */
/* CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,   */
/* EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,     */
/* PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR      */
/* PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY     */
/* OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT            */
/* (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE   */
/* OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.    */
/* * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * */

#include "athd.h"

/* private protos */
static apr_status_t _athd_engine_run_reqs(athd_engine *engine);
static apr_status_t _athd_engine_run_req(athd_request *req);
static void _athd_engine_thread_cleanup(void *ctx);

/* private data */
static ath_list *_retired_conns;

/* end privates */

#define ETHR "Engine Thread: [%s]"

void *
athd_engine_thread(void *ctx)
{
    int rv = APR_SUCCESS;
    athd_engine *engine = ctx;
    char *strerr = NULL;
    apr_time_t mytick = XTH_SEC(10); /* 1/10 sec */
    apr_time_t retrywait = (mytick * 10);
    apr_time_t retryfactor = 2;
    int laststate, lasttype;
    int failures = 0;
    struct sigaction action = { NULL } ;
    ASSERT(engine);

    pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, &laststate);
    pthread_setcanceltype(PTHREAD_CANCEL_DEFERRED, &lasttype);
    pthread_cleanup_push((void *)_athd_engine_thread_cleanup, (void *)engine);

    action.sa_handler = athd_connection_reset;
    sigaction(SIGPIPE, &action, 0);

    athd_log(ATHD_INFO, ETHR " Enter", engine->name);
    while (!engine->dienow)
      {
	  pthread_testcancel();
	  while (!engine->dienow)
	    {
		pthread_testcancel();
		rv = _athd_engine_run_reqs(engine);
		if (ATHD_STATUS_IS_FATAL(rv))
		  {
		      /* refusing prevents reqs from being pushed to queue */
		      /* store reason */
		      engine->refusing = rv;
		      athd_log_err(ATHD_CRIT, rv, ETHR " Fatal error: ",
					 engine->name);
		      engine->dienow = 1;
		  }
		else if (! APR_STATUS_IS_SUCCESS(rv))
		  {
		      engine->refusing = rv;
		      apr_sleep(retrywait);
		      retrywait *= retryfactor;
		  }
		else
		  {
		      retrywait = XTH_SEC(100);
		      engine->refusing = 0;
		  }
		if (!engine->dienow)
		    apr_sleep(mytick);
	    }
      }
    pthread_cleanup_pop(1);
}

void
athd_engine_thread_shared_setup(apr_pool_t *pglobal)
{
    /* nothing */
}


/* privates below */

/* runs all requests in an engine's request queue 
   each engine gets 1 thread and 1 connection to prevent overloading
   of remote server. should be adequate, even up to 1000 member farms.
   todo: config arg to use >1 conns per engine?
*/
static apr_status_t
_athd_engine_run_reqs(athd_engine *engine)
{
    int rv = APR_SUCCESS;
    athd_request *req = NULL;
    ath_node *list_ctx;
    /* all relevant sync handlerd in list code  */
    while ((athd_engine_req_qsize(engine) > 0)
	   && (!rv) && (!engine->dienow)
	   && ((req = athd_engine_shift_req(engine)) != NULL))
      {
	  ASSERT(req);
	  if ((apr_time_now()) > req->expires)
	    {
		athd_log(ATHD_WARN, "Job: [%s] Request Expired: [%s]",
			 req->job ? req->job->name : "(internal)", req->url);
		athd_req_destroy(&req);
	    }
	  else
	    {
		rv = _athd_engine_run_req(req);
		if (rv)
		  {
		      athd_log_err(ATHD_WARN, rv, 
				   "Job: [%s] Request: [%s] (canceled) Error: ",
				   req->job ? req->job->name : "(internal)", 
				   req->url);
		      athd_req_destroy(&req);
		  }
		else if (req->job)
		    athd_job_push_response(req);
	    }
	  /* be nice: 1/20th sec * slow factor */
	  if (!engine->dienow)
	      apr_sleep((XTH_SEC(20)) * engine->slow_server_factor); 
      }
    return rv;
}

static apr_status_t
_athd_engine_run_req(athd_request *req)
{
    int rv = APR_SUCCESS;
    serf_filter_t *filter;
    ASSERT(req);
    ASSERT(req->pool);
    ASSERT(req->url);
    ASSERT(req->engine);

    pthread_testcancel();
    req->tries++;

    if (req->tries > 1)
	athd_log(ATHD_DEBUG, "Request: [%s] Retry: [%d/%d] Engine: [%s]", 
		 req->url, req->tries - 1, ATHD_REQ_GIVEUP,
		 req->engine->name);

    athd_log(ATHD_DEVEL, "Job: [%s] Request: [%s]", 
	     req->job ? req->job->name : "(internal)", req->url);

    if ((!req->engine->conn_current)
	|| (athd_connection_is_retired(req->engine->conn_current)))
	rv = athd_engine_new_connection(req->engine);

    req->conn = req->engine->conn_current;

    if (rv)
	athd_log_err(ATHD_DEBUG, rv, 
		     "Request: [%s] Connection error: ", req->url);
    else if (!req->conn)
	rv = APR_EINVALSOCK;
    else 
      {
	  rv = athd_connection_do_serf(req);

	  if ((!rv) && (!req->engine->keep_alive))
	      athd_connection_retire(req->conn);    
	  else if (rv)
	    {
		if (rv == ATHD_BAD_KEEPALIVE)
		    athd_log(ATHD_NOTE,
			     ETHR " KeepAlive timeout (code %d), "
			     "consider turning KeepAlive off "
			     "or lowering job frequency value",
			     req->engine->name, ATHD_BAD_KEEPALIVE); 
		/* connection is broken */
		athd_connection_retire(req->conn);
		/* i'm not using it anymore */
		athd_connection_release(&req->engine->conn_current);
		req->conn = NULL;
		/* successful conns are released in athd_req_handler_out */
		if (req->tries < ATHD_REQ_GIVEUP)
		  {
		      /* be nice */
		      apr_sleep((XTH_SEC(20)) 
				* req->engine->slow_server_factor);
		      rv = _athd_engine_run_req(req);
		  }
		else if ((ATHD_REQ_GIVEUP > 1) 
			 && (req->tries >= ATHD_REQ_GIVEUP))
		  {
		      athd_log(ATHD_NOTE, 
			       "Request: [%s] Giving up after [%d] tries",
			       req->url, req->tries);
		  }
	    }
      }
    return rv;
}

static void
_athd_engine_thread_cleanup(void *ctx)
{
    athd_engine *engine = (athd_engine *)ctx;
    athd_log(ATHD_INFO, ETHR " Cleanup/Exit", engine->name); 
}
